package com.up.interview.queue;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author: liuchengjun
 * @title: BlockingQueueExamples
 * @projectName: 海联技术中台V2.0
 * @depart 软件事业部
 * @description:
 * @date: 2025/10/21/周二 9:46 (星期二)
 * @since V2.0
 */
public class BlockingQueueExamples {
    // ========== 1. ArrayBlockingQueue：有界队列，适合固定容量场景 - 订单处理系统 ==========
    static class OrderProcessingSystem {
        private final BlockingQueue<Order> orderQueue = new ArrayBlockingQueue<>(50);
        private final AtomicInteger orderIdGenerator = new AtomicInteger(1);

        static class Order {
            int id;
            String product;
            double amount;

            Order(int id, String product, double amount) {
                this.id = id;
                this.product = product;
                this.amount = amount;
            }

            @Override
            public String toString() {
                return "订单[" + id + "] " + product + " ¥" + amount;
            }
        }

        // 接收订单（生产者）
        public void receiveOrder(String product, double amount) throws InterruptedException {
            Order order = new Order(orderIdGenerator.getAndIncrement(), product, amount);
            System.out.println("接收订单: " + order);
            orderQueue.put(order); // 队列满时会阻塞
        }

        // 处理订单（消费者）
        public void startProcessing() {
            new Thread(() -> {
                try {
                    while (true) {
                        Order order = orderQueue.take(); // 队列空时会阻塞
                        System.out.println("处理订单: " + order);
                        Thread.sleep(1000); // 模拟处理时间
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }, "订单处理线程").start();
        }
    }

    // ========== 2. LinkedBlockingQueue：有界/无界，适合高吞吐场景 - 日志异步写入 ==========
    static class AsyncLogSystem {
        private final BlockingQueue<LogMessage> logQueue = new LinkedBlockingQueue<>(1000);

        static class LogMessage {
            String level;
            String message;
            long timestamp;

            LogMessage(String level, String message) {
                this.level = level;
                this.message = message;
                this.timestamp = System.currentTimeMillis();
            }

            @Override
            public String toString() {
                return "[" + level + "] " + timestamp + " - " + message;
            }
        }

        public void log(String level, String message) {
            try {
                logQueue.offer(new LogMessage(level, message), 100, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }

        public void startLogWriter() {
            new Thread(() -> {
                try {
                    while (true) {
                        LogMessage log = logQueue.take();
                        // 模拟写入磁盘
                        System.out.println("写入日志: " + log);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }, "日志写入线程").start();
        }
    }
    //重写compareTo对队列元素进行排序
    // ========== 3. PriorityBlockingQueue - 任务优先级调度 ==========
    static class TaskScheduler {
        private final BlockingQueue<Task> taskQueue = new PriorityBlockingQueue<>();

        static class Task implements Comparable<Task> {
            String name;
            int priority; // 数字越小优先级越高

            Task(String name, int priority) {
                this.name = name;
                this.priority = priority;
            }

            @Override
            public int compareTo(Task other) {
                return Integer.compare(this.priority, other.priority);
            }

            @Override
            public String toString() {
                return "任务[" + name + ", 优先级:" + priority + "]";
            }
        }

        public void submitTask(String name, int priority) throws InterruptedException {
            Task task = new Task(name, priority);
            System.out.println("提交任务: " + task);
            taskQueue.put(task);
        }

        public void startExecution() {
            new Thread(() -> {
                try {
                    while (true) {
                        Task task = taskQueue.take();
                        System.out.println("执行任务: " + task);
                        Thread.sleep(500);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }, "任务执行线程").start();
        }
    }

    // ========== 4. DelayQueue - 延迟任务调度：时间长短进行排序 ==========
    static class DelayedTaskScheduler {
        private final BlockingQueue<DelayedTask> delayQueue = new DelayQueue<>();

        static class DelayedTask implements Delayed {
            String name;
            long executeTime; // 执行时间戳

            DelayedTask(String name, long delayMillis) {
                this.name = name;
                this.executeTime = System.currentTimeMillis() + delayMillis;
            }

            @Override
            public long getDelay(TimeUnit unit) {
                long diff = executeTime - System.currentTimeMillis();
                return unit.convert(diff, TimeUnit.MILLISECONDS);
            }

            @Override
            public int compareTo(Delayed o) {
                return Long.compare(this.executeTime, ((DelayedTask) o).executeTime);
            }

            @Override
            public String toString() {
                return "延迟任务[" + name + "]";
            }
        }

        public void scheduleTask(String name, long delaySeconds) throws InterruptedException {
            DelayedTask task = new DelayedTask(name, delaySeconds * 1000);
            System.out.println("调度任务: " + task + ", 延迟" + delaySeconds + "秒");
            delayQueue.put(task);
        }

        public void startExecution() {
            new Thread(() -> {
                try {
                    while (true) {
                        DelayedTask task = delayQueue.take(); // 时间未到会阻塞
                        System.out.println("执行延迟任务: " + task);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }, "延迟任务执行线程").start();
        }
    }

    // ========== 5. SynchronousQueue - 数据直接传递 ==========
    static class DataTransferSystem {
        private final BlockingQueue<String> transferQueue = new SynchronousQueue<>();

        // 数据发送者
        public void sender() {
            new Thread(() -> {
                try {
                    for (int i = 1; i <= 5; i++) {
                        String data = "数据包-" + i;
                        System.out.println("发送方准备发送: " + data);
                        transferQueue.put(data); // 必须等待接收方取走
                        System.out.println("发送方已发送: " + data);
                        Thread.sleep(1000);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }, "发送线程").start();
        }

        // 数据接收者
        public void receiver() {
            new Thread(() -> {
                try {
                    for (int i = 1; i <= 5; i++) {
                        Thread.sleep(2000); // 模拟处理延迟
                        String data = transferQueue.take(); // 等待发送方
                        System.out.println("接收方收到: " + data);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }, "接收线程").start();
        }
    }


    public static void main(String[] args) {
        try {
            System.out.println("\n========== 3. PriorityBlockingQueue - 优先级任务 ==========");
            TaskScheduler scheduler = new TaskScheduler();
            scheduler.startExecution();
            scheduler.submitTask("普通任务A", 5);
            scheduler.submitTask("紧急任务B", 1);
            scheduler.submitTask("普通任务C", 5);
            scheduler.submitTask("高优先级任务D", 2);
            Thread.sleep(3000);


            System.out.println("\n========== 4. DelayQueue - 延迟任务 ==========");
            DelayedTaskScheduler delayScheduler = new DelayedTaskScheduler();
            delayScheduler.startExecution();
            delayScheduler.scheduleTask("5秒后执行", 5);
            delayScheduler.scheduleTask("2秒后执行", 2);
            delayScheduler.scheduleTask("8秒后执行", 8);
            Thread.sleep(10000);


            System.out.println("\n========== 5. SynchronousQueue - 直接传递 ==========");
            DataTransferSystem transferSystem = new DataTransferSystem();
            transferSystem.sender();
            transferSystem.receiver();
            Thread.sleep(12000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }



}
